package com.liu.kafka.control;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * 集成KafkaTemplate
 */
@Slf4j
@RestController
@RequestMapping("/kafka")
public class ProducerController {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    private final String topic = "first";

    /**
     * 没有指定分区，但有key的情况，将key的hash与分区数取余计算分区
     * 没有分区和key时，随机选择分区
     * 可自定义分区
     * @param msg
     */
    @GetMapping("/{msg}")
    @Transactional(rollbackFor = RuntimeException.class)// kafka事务开启  acks=-1   transaction-id-prefix: test
    public String sendMessage(@PathVariable("msg") String msg) {
        //随机
        kafkaTemplate.send(topic,"事务1-->"+ msg);
        if("error".equals(msg)){
            throw new RuntimeException();
        }
        kafkaTemplate.send(topic,"事务2-->"+ msg);
        return "发送成功："+msg;
    }

    //同步发送
    @GetMapping("/sync")
    public void sync( String msg) throws Exception {

        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, JSON.toJSONString(msg));
        //注意，可以设置等待时间，超出后，不再等候结果
        SendResult<String, String> result = future.get(3, TimeUnit.SECONDS);
        log.info("send result:{}", result.getProducerRecord().value());
    }


    /**
     *  指定分区发送,不管你key是什么，到同一个分区
     * @param key
     */
    @GetMapping("/partitionSend/{key}")
    public void setPartition(@PathVariable("key") String key) {
        kafkaTemplate.send(topic, 0, key, "key=" + key + "，msg=指定0号分区");
    }


    /**
     * 指定key发送，不指定分区,根据key做hash，相同的key到同一个分区
     * @param key
     */
    @GetMapping("/keySend/{key}")
    public void setKey(@PathVariable("key") String key) {
        kafkaTemplate.send(topic, key, "key=" + key + "，msg=根据key做hash");
    }

}
